package com.Connection;

import com.sink.IISSink;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;

@AllArgsConstructor
@NoArgsConstructor
@Data

/**
*@description：JDBC工具类，生成各类数据库的连接对象，用于sink的创建
*@Author: 谢秉均
*@date: 2024/12/31 16:44
*/

public class JDBCConnectionTools implements Serializable {

    private String sqltype; //sink输出的jdbc的类型

    private static final Logger logger = LoggerFactory.getLogger(JDBCConnectionTools.class);


    public JdbcConnectionOptions GetMssqlConnect(String user, String password, String DBip, String databaseName){

        JdbcConnectionOptions connectionOptions = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()    //sql server数据库sink
                .withConnectionCheckTimeoutSeconds(10) //单位s; 重试连接的间隔,上面我们设置最多重试3次,每次间隔10s
                .withUsername(user)
                .withPassword(password)
                .withUrl("jdbc:sqlserver://"+DBip+":1433;databaseName="+databaseName)
                .withDriverName("com.microsoft.sqlserver.jdbc.SQLServerDriver")
                .build();
        return connectionOptions;
    }

    public JdbcConnectionOptions GetMysqlConnect(String user, String password, String DBip, String databaseName){

        JdbcConnectionOptions connectionOptions = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()    //sql server数据库sink
                // mysql 默认8小时不使用连接就主动断开连接
                .withConnectionCheckTimeoutSeconds(10) //单位s; 重试连接的间隔,上面我们设置最多重试3次,每次间隔10s
                .withUsername(user)
                .withPassword(password)
                .withUrl("jdbc:mysql://"+DBip+":3306/"+databaseName+"?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=GMT%2B8")
                .build();
        return connectionOptions;
    }


    
    /**
    *@description：根据类初始化时确定的数据库类型生成对应的jdbc连接，用于sink的创建
    *@Param:
    *@return:
    *@Author: 谢秉均
    *@date: 2024/12/31 16:42
    */
    
    public JdbcConnectionOptions GetConnect(String user, String password, String DBip, String databaseName) throws Exception {
        if (this.sqltype.equals("mysql")){
            return  this.GetMysqlConnect(user, password, DBip,databaseName);
        }if(this.sqltype.equals("mssql")){
            return  this.GetMssqlConnect(user, password, DBip,databaseName);
        }else {
            logger.error("类型"+this.sqltype+"暂不支持或sqltype未进行初始化");
            throw new Exception("类型"+this.sqltype+"暂不支持或sqltype未进行初始化");
        }

    }


}
